package io.shdxhl.simplerpc.server.boot.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.NettyRuntime;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.concurrent.UnorderedThreadPoolEventExecutor;
import io.shdxhl.simplerpc.code.config.SimpleConfiguration;
import io.shdxhl.simplerpc.code.netty.RpcServer;
import io.shdxhl.simplerpc.code.netty.codec.FrameDecoder;
import io.shdxhl.simplerpc.code.netty.codec.FrameEncoder;
import io.shdxhl.simplerpc.code.netty.codec.RpcRequestDecoder;
import io.shdxhl.simplerpc.code.netty.codec.RpcResponseEncoder;
import io.shdxhl.simplerpc.code.netty.handler.RpcRequestHandler;
import lombok.extern.slf4j.Slf4j;

import javax.annotation.Resource;
import java.net.InetSocketAddress;

@Slf4j
public class NettyServer implements RpcServer {

    @Resource
    private SimpleConfiguration configuration;

    @Override
    public void start() {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("netty-boss"));
        EventLoopGroup workerGroup = new NioEventLoopGroup(0, new DefaultThreadFactory("netty-worker"));
        EventExecutorGroup businessGroup = new UnorderedThreadPoolEventExecutor(NettyRuntime.availableProcessors() *2, new DefaultThreadFactory("netty-business"));

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler())
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            //add encoder
                            pipeline.addLast("frameEncode", new FrameEncoder());
                            pipeline.addLast("serialize", new RpcResponseEncoder());
                            //add decoder
                            pipeline.addLast("frameDecoder",  new FrameDecoder());
                            //add Handler
                            pipeline.addLast("deserialize", new RpcRequestDecoder());
                            pipeline.addLast("requestHandler", new RpcRequestHandler());
                        }
                    });
            ChannelFuture channelFuture = bootstrap.bind(new InetSocketAddress(configuration.getPort())).sync();

            //添加CHANEL Future关闭监听，释放当前线程
            channelFuture.channel().closeFuture().addListener(new ChannelFutureListener() {

                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    bossGroup.shutdownGracefully();
                    workerGroup.shutdownGracefully();
                    businessGroup.shutdownGracefully();
                }
            });

        } catch (InterruptedException e) {
            log.error("NettyServer start error", e);
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            businessGroup.shutdownGracefully();
        }
    }
}
